-module(rtps_history_cache).

%%% @doc The HistoryCache is part of the interface between the user
%%% application and RTPS and is used to temporarily store and manage
%%% sets of changes to data-objects. See 8.2.2. On the Writer side it
%%% contains the history of the changes to data-objects made by the
%%% Writer. On the Reader side it contains the history of the changes
%%% to data-objects made by the matched RTPS Writer endpoints.

-behaviour(gen_server).

-include("rtps.hrl").
-include("rtps_history_cache.hrl").

%%%===================================================================
%%% Types
%%%===================================================================
-export_type([cache_change/0, change_kind/0, data_value/0, instance_handle/0]).
-type cache_change() :: #cache_change{}.
-type change_kind() :: alive | not_alive_unregistered | not_alive_disposed.
-type data_value() :: binary() | undefined. % term().
-type instance_handle() :: term() | undefined.

%% API
-export([start_link/1, add_sup_handler/3, delete_handler/3, new/1,
	 available_changes/1, add_change/2, get_change/3, remove_change/2, remove_change/3,
	 remove_changes/2, get_seq_num_min/2, get_seq_num_max/2, get_seq_num_min_max/2]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	 terminate/2, code_change/3]).

-record(hcc, {writer_guid :: rtps:guid(),
	      storage :: cache_change() | ets:tid(),
	      first_sn :: rtps:sequence_number() | undefined,
	      last_sn :: rtps:sequence_number() | undefined}).
-record(data, {hccs :: [#hcc{}], evt_mgr :: pid(),
	       kind :: rtps:history_kind(),
	       depth :: rtps:history_depth() | undefined,
	       durability :: rtps:history_durability(),
	       reliability_level :: rtps:reliability_level()}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Opts) ->
    gen_server:start_link(?MODULE, [Opts], []).
			       
-spec new(Opts) -> pid() when
      Opts :: [proplists:property()].
%% @doc This operation creates a new RTPS HistoryCache. The
%% newly-created history cache is initialized with an empty list of
%% changes.
new(Opts) ->
    {ok, Pid} = start_link(Opts),
    Pid.

-spec available_changes(Pid) -> [Change] when
      Pid :: pid(),
      Change :: cache_change().
%% @doc retruns all changes from the HistoryCahce.
available_changes(Pid) ->
    gen_server:call(Pid, available_changes).

-spec add_change(Pid, Change | [Change]) -> ok | {error, Reason} when
      Pid :: pid(),
      Change :: cache_change(),
      Reason :: already_present | term().
%% @doc This operation inserts the CacheChange `Change' or a list of
%% `Changes' into the HistoryCache. Adding an already existing
%% CacheChange will result in an `already_present' error.
add_change(Pid, Changes) when is_list(Changes) ->
    gen_server:call(Pid, {add_change, Changes});
add_change(Pid, Change) ->
    add_change(Pid, [Change]).

-spec get_change(Pid, Writer_guid, Sequence_number) -> Change | false when
      Pid :: pid(),
      Writer_guid :: rtps:guid(), 
      Sequence_number :: rtps:sequence_number(),
      Change :: cache_change().
%% @doc Fetch a CacheChange Change from the HistoryCache using the
%% Writer's GUID `Writer_guid' and the SequenceNumber
%% `Sequence_number' as keys.
get_change(Pid, Writer_guid, Sequence_number) ->
    gen_server:call(Pid, {get_change, Writer_guid, Sequence_number}).

-spec remove_change(Pid, Change) -> ok when
      Pid :: pid(),
      Change :: cache_change().
%% @doc This operation indicates that a previously-added CacheChange
%% has become irrelevant and the details regarding the CacheChange
%% need not be maintained in the HistoryCache.
%% @see rtps_history_cache:remove_change/3.
remove_change(Pid, #cache_change{writer_guid = Writer_guid, sequence_number = Sequence_number}) ->
    remove_change(Pid, Writer_guid, Sequence_number).

-spec remove_change(Pid, Writer_guid, Sequence_number) -> ok when
      Pid :: pid(),
      Writer_guid :: rtps:guid(), 
      Sequence_number :: rtps:sequence_number().
%% @doc Remove a CacheChange with keys `Writer_guid', `Sequence_number' from the HistoryCache,
remove_change(Pid, Writer_guid, Sequence_number) ->
    gen_server:call(Pid, {remove_change, Writer_guid, Sequence_number}).

-spec remove_changes(Pid, Guid) -> ok when
      Pid :: pid(),
      Guid :: rtps:guid().
%% @doc NOT IN THE SPECS Removes all the changes associated with the `Guid'.
remove_changes(Pid, Guid) ->
    gen_server:call(Pid, {remove_changes, Guid}).

-spec get_seq_num_min(Pid, Writer_guid) -> rtps:sequence_number() | undefined when
      Pid :: pid(),
      Writer_guid :: rtps:guid().      
%% @doc This operation retrieves the smallest value of the
%% CacheChange::sequenceNumber attribute among the CacheChange stored
%% in the HistoryCache. Returns `undefined' if the HistoryCache is
%% empty.
%%
%% NB: In the specifications, the writer's guid is omitted as an
%% argument but it is needed to make this function meaningful. A
%% history cache may contain changes from more than one writer, so a
%% writer guid is required.
get_seq_num_min(Pid, Writer_guid) ->
    gen_server:call(Pid, {get_seq_num_min, Writer_guid}).
			       
-spec get_seq_num_max(Pid, Writer_guid) -> rtps:sequence_number() | undefined when
      Pid :: pid(),
      Writer_guid :: rtps:guid().      
%% @doc This operation retrieves the largest value of the
%% CacheChange::sequenceNumber attribute among the CacheChange stored
%% in the HistoryCache. Returns `undefined' if the HistoryCache is
%% empty. @see get_seq_num_min
get_seq_num_max(Pid, Writer_guid) ->
    gen_server:call(Pid, {get_seq_num_max, Writer_guid}).

-spec get_seq_num_min_max(Pid, Writer_guid) -> {First_sn, Last_sn} | undefined when
      Pid :: pid(),
      Writer_guid :: rtps:guid(),
      First_sn :: rtps:sequence_number(),
      Last_sn :: rtps:sequence_number().
%% @doc NOT IN THE SPECS This operation retrieves the smallest and
%% largest value of the CacheChange::sequenceNumber attribute among
%% the CacheChange stored in the HistoryCache. Returns `undefined' if
%% the HistoryCache is empty. @see get_seq_num_min
get_seq_num_min_max(Pid, Writer_guid) ->
    gen_server:call(Pid, {get_seq_num_min_max, Writer_guid}).

-spec add_sup_handler(Pid, Module, Args) -> Result when
      Pid :: pid(),
      Module :: atom(),
      Args :: term(),
      Result :: ok | {'EXIT',Reason} | term(),
      Reason :: term().
%% @doc NOT IN THE SPECS Adds a new event supervised handler to the
%% history cache's event manager.
add_sup_handler(Pid, Module, Args) ->
    Evt_mgr = gen_server:call(Pid, evt_mgr),
    Handler = {Module, self()},
    gen_event:add_sup_handler(Evt_mgr, Handler, [self() | Args]).

-spec delete_handler(Pid, Module, Args) -> Result when
      Pid :: pid(),
      Module :: atom(),
      Args :: term(),
      Result :: ok | {'EXIT',Reason} | term(),
      Reason :: term().
%% @doc NOT IN THE SPECS Deletes an event handler from the history
%% cache's event manager.
delete_handler(Pid, Module, Args) ->
    Evt_mgr = gen_server:call(Pid, evt_mgr),
    Handler = {Module, self()},
    gen_event:delete_handler(Evt_mgr, Handler, Args).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%% @private
init([Opts]) ->
    {ok, Evt_mgr} = gen_event:start_link(),
    Kind = proplists:get_value(history_kind, Opts, keep_all),
    Depth = proplists:get_value(history_depth, Opts),
    Durability = proplists:get_value(history_durability, Opts, volatile),
    case proplists:get_value(reliability_level, Opts) of
	undefined ->
	    {stop, {badopt, reliability_level}};
	_ when Kind =:= keep_last, Depth =:= undefined ->
	    {stop, {badopt, depth}};
	Reliability_level ->
	    {ok, #data{hccs = [], evt_mgr = Evt_mgr, kind = Kind, depth = Depth,
		       durability = Durability, reliability_level = Reliability_level}}
    end.

%% @private

%% TODO: The list of changes must share the same writer GUID! The GUID
%% from the first Change is used while processing all other Changes in
%% the list. This is not a problem as far as the Changes stem from
%% within `rtps' because these lists will always be correct, but may
%% be a problem in case an external user application adds CacheChanges
%% to the HistoryCache. The external API should implement proper
%% checks.
handle_call({add_change, [#cache_change{writer_guid = Writer_guid} | _] = Changes}, From,
	    #data{hccs = Hccs, evt_mgr = Evt_mgr, kind = Kind, depth = Depth,
		  durability = Durability, reliability_level = Reliability_level} = Data) ->
    case add_change(Kind, Depth, Durability, Reliability_level, Writer_guid, Changes, Hccs) of
	{ok, Hccs1} ->
	    gen_event:notify(Evt_mgr, {add_change, From, Changes}),
	    {reply, ok, Data#data{hccs = Hccs1}};
	Error ->
	    {reply, Error, Data}
    end;
handle_call({remove_change, Writer_guid, Sequence_number}, From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level, evt_mgr = Evt_mgr} = Data) ->
    {Hccs1, Seq_nums} = remove_change(Kind, Depth, Durability, Reliability_level, Writer_guid, Sequence_number, Hccs),
    case Seq_nums of
	[] ->
	    ok;
	_ ->
	    gen_event:notify(Evt_mgr, {remove_change, From, Writer_guid, Seq_nums})
    end,
    {reply, ok, Data#data{hccs = Hccs1}};
handle_call({remove_changes, Writer_guid}, From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level, evt_mgr = Evt_mgr} = Data) ->
    {Hccs1, Seq_nums} = remove_changes(Kind, Depth, Durability, Reliability_level, Writer_guid, Hccs),
    case Seq_nums of
	[] ->
	    ok;
	_ ->
	    gen_event:notify(Evt_mgr, {remove_change, From, Writer_guid, Seq_nums})
    end,
    {reply, ok, Data#data{hccs = Hccs1}};
handle_call({get_change, Writer_guid, Sequence_number}, _From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level} = Data) ->
    Reply = get_change(Kind, Depth, Durability, Reliability_level, Writer_guid, Sequence_number, Hccs),
    {reply, Reply, Data};
handle_call({get_changes, Writer_guid}, _From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level} = Data) ->
    Reply = get_changes(Kind, Depth, Durability, Reliability_level, Writer_guid, Hccs),
    {reply, Reply, Data};
handle_call({get_seq_num_min, Writer_guid}, _From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level} = Data) ->
    Reply = get_seq_num_min(Kind, Depth, Durability, Reliability_level, Writer_guid, Hccs),
    {reply, Reply, Data};
handle_call({get_seq_num_max, Writer_guid}, _From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level} = Data) ->
    Reply = get_seq_num_max(Kind, Depth, Durability, Reliability_level, Writer_guid, Hccs),
    {reply, Reply, Data};
handle_call({get_seq_num_min_max, Writer_guid}, _From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level} = Data) ->
    case get_seq_num_min(Kind, Depth, Durability, Reliability_level, Writer_guid, Hccs) of
	undefined ->
	    {reply, undefined, Data};
	First_sn ->
	    Last_sn = get_seq_num_max(Kind, Depth, Durability, Reliability_level, Writer_guid, Hccs),
	    {reply, {First_sn, Last_sn}, Data}
    end;
handle_call(available_changes, _From,
	    #data{hccs = Hccs, kind = Kind, depth = Depth, durability = Durability,
		  reliability_level = Reliability_level} = Data) ->
    Reply = available_changes(Kind, Depth, Durability, Reliability_level, Hccs),
    {reply, Reply, Data};
handle_call(evt_mgr, _From, #data{evt_mgr = Evt_mgr} = Data) ->
    Reply = Evt_mgr,
    {reply, Reply, Data}.

%% @private
handle_cast(_Msg, Data) ->
    {noreply, Data}.

%% @private
handle_info(_Info, Data) ->
    {noreply, Data}.

%% @private
terminate(_Reason, _Data) ->
    ok.

%% @private
code_change(_OldVsn, Data, _Extra) ->
    {ok, Data}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

add_change(keep_last, 1, _Durability, _Reliability_level, Writer_guid, Changes, Hccs) ->
    #cache_change{sequence_number = Seq_num} = Change = lists:last(Changes),
    Hcc1 = #hcc{writer_guid = Writer_guid, storage = Change, first_sn = Seq_num, last_sn = Seq_num},
    {ok, lists:keystore(Writer_guid, #hcc.writer_guid, Hccs, Hcc1)};

    %% TODO: with the best-effort stateless reader, the sequence
    %% number is NOT guaranteed to be higher than the previous
    %% time. Strictly speaking in the situation that a cache change
    %% with the same sequence number arrives, that cache change is
    %% acceptable if the 'content' is the same (serialized data, key)
    %% and only the timestamp should be actualized. Currently we do
    %% not perform any checks on content but just replace the cache
    %% change.

    %% TODO: Why bother with error checking, when we know the writer
    %% will (should) always come up with a higher sequence number!?
    %% case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
    %% 	false ->
    %% 	    {ok, [Hcc1 | Hccs]};
    %% 	#hcc{last_sn = Last_sn} ->
    %% 	    case Last_sn < Seq_num of
    %% 		true ->
    %% 		    {ok, lists:keystore(Writer_guid, #hcc.writer_guid, Hccs, Hcc1)};
    %% 		false ->
    %% 		    {error, already_present}
    %% 	    end
    %% end;
add_change(_Kind, _Depth, _Durability, _Reliability_level, Writer_guid, Changes, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
    	#hcc{storage = Table} ->
    	    %% TODO: Again, why bother with the error checking? Use
    	    %% insert_new and NOT insert.
    	    case ets:insert_new(Table, Changes) of
    		true ->
		    {ok, Hccs};
    		false ->
    		    %% Will fail if a change is already present for
    		    %% the sequence number. In case a list of changes
    		    %% is inserted, the operation as a whole will fail
    		    %% if one of the changes is already present.
    		    {error, already_present}
    	    end;
    	false ->
    	    Table = ets:new(history_cache, [ordered_set,
    					    %% private,
    					    {keypos, #cache_change.sequence_number}]),
    	    ets:insert(Table, Changes),
    	    Hcc = #hcc{writer_guid = Writer_guid, storage = Table},
    	    {ok, [Hcc | Hccs]}
    end.
    
get_change(keep_last, 1, _Durability, _Reliability_level, Writer_guid, Sequence_number, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{storage = Change, last_sn = Sequence_number} ->
	    Change;
	_ ->
	    false
    end;
get_change(_Kind, _Depth, _Durability, _Reliability_level, Writer_guid, Sequence_number, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{storage = Table} ->
	    case ets:lookup(Table, Sequence_number) of
		[Change] ->
		    Change;
		_ ->
		    false
	    end;
	false ->
	    false
    end.

get_changes(keep_last, 1, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{storage = Change} ->
	    [Change];
	_ ->
	    []
    end;
get_changes(_Kind, _Depth, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{storage = Table} ->
	    %% TODO: This is acording to the specs, but it is a
	    %% potential cause for crashes if the table is large and
	    %% memory is limited and totally inefficient for large
	    %% tables.
	    [N || #cache_change{sequence_number = N} <- ets:tab2list(Table)];
	false ->
	    []
    end.

get_seq_num_min(keep_last, 1, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{first_sn = First_sn} ->
	    First_sn;
	_ ->
	    undefined
    end;
get_seq_num_min(_Kind, _Depth, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{storage = Table} ->
	    case ets:first(Table) of
		'$end_of_table' ->
		    undefined;
		First_sn ->
		    First_sn
	    end;
	false ->
	    undefined
    end.

get_seq_num_max(keep_last, 1, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{last_sn = Last_sn} ->
	    Last_sn;
	_ ->
	    undefined
    end;
get_seq_num_max(_Kind, _Depth, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{storage = Table} ->
	    case ets:last(Table) of
		'$end_of_table' ->
		    undefined;
		First_sn ->
		    First_sn
	    end;
	false ->
	    undefined
    end.

available_changes(keep_last, 1, _Durability, _Reliability_level, Hccs) ->
    [Change || #hcc{storage = Change} <- Hccs];
available_changes(_Kind, _Depth, _Durability, _Reliability_level, Hccs) ->
    lists:flatten([ets:tab2list(Table) || #hcc{storage = Table} <- Hccs]).
    
remove_change(_Kind, _Depth, Durability, _Reliability_level, _Writer_guid, _Sequence_number, Hccs)
  when Durability =:= transient_local; Durability =:= transient ->
    {Hccs, []};
remove_change(keep_last, 1, _Durability, _Reliability_level, Writer_guid, Sequence_number, Hccs) ->
    case lists:keytake(Writer_guid, #hcc.writer_guid, Hccs) of
	{value, #hcc{storage = #cache_change{sequence_number = Sequence_number}}, Hccs1} ->
	    {Hccs1, [Sequence_number]};
	_ ->
	    {Hccs, []}
    end;
remove_change(_Kind, _Depth, _Durability, _Reliability_level, Writer_guid, Sequence_number, Hccs) ->
    %% TODO: Should we delete the table when it is empty? Adds
    %% overhead but does free resources in situations that the history
    %% cache is flooded with writer guids. This is unlikely, so let's
    %% keep it simple by not checking on empty table and removing them.
    case lists:keyfind(Writer_guid, #hcc.writer_guid, Hccs) of
	#hcc{storage = Table} ->
	    ets:delete(Table, Sequence_number),
	    {Hccs, [Sequence_number]};
	false ->
	    {Hccs, []}
    end.

remove_changes(_Kind, _Depth, Durability, _Reliability_level, _Writer_guid, Hccs)
  when Durability =:= transient_local; Durability =:= transient ->
    {Hccs, []};
remove_changes(keep_last, 1, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keytake(Writer_guid, #hcc.writer_guid, Hccs) of
	{value, #hcc{storage = #cache_change{sequence_number = Sequence_number}}, Hccs1} ->
	    {Hccs1, [Sequence_number]};
	false ->
	    {Hccs, []}
    end;
remove_changes(_Kind, _Depth, _Durability, _Reliability_level, Writer_guid, Hccs) ->
    case lists:keytake(Writer_guid, #hcc.writer_guid, Hccs) of
	{value, #hcc{storage = Table}, Hccs1} ->
	    Seq_nums = [Seq_num || #cache_change{sequence_number = Seq_num} <- ets:tab2list(Table)],
	    ets:delete(Table),
	    {Hccs1, Seq_nums};
	false ->
	    {Hccs, []}
    end.
